/*-------------------------------------------------------------------------
 *
 * tqueue.c
 *	  使用 shm_mq 在并行后端之间发送和接收元组
 *
 * 一个类型为 DestTupleQueue 的 DestReceiver，底层是 TQueueDestReceiver，
 * 将元组从执行器写入 shm_mq。
 *
 * 一个 TupleQueueReader 从 shm_mq 中读取元组并返回这些元组。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/executor/tqueue.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/htup_details.h"
#include "executor/tqueue.h"

/*
 * DestReceiver对象的私有内容
 *
 * queue是指向由DestReceiver的调用者提供的数据的指针。
 */
typedef struct TQueueDestReceiver
{
	DestReceiver pub;			/* 公共字段 */
	shm_mq_handle *queue;		/* shm_mq发送到 */
} TQueueDestReceiver;

/*
 * TupleQueueReader对象的私有内容
 *
 * queue是指向由读取者的调用者提供的数据的指针。
 *
 * "typedef struct TupleQueueReader TupleQueueReader" 在tqueue.h中
 */
struct TupleQueueReader
{
	shm_mq_handle *queue;		/* shm_mq接收自 */
};

/*
 * 从查询接收一个元组，并将其发送到指定的shm_mq。
 *
 * 如果成功则返回true，如果shm_mq已被分离则返回false。
 */
static bool fc_tqueueReceiveSlot(TupleTableSlot *fc_slot, DestReceiver *fc_self)
{
	TQueueDestReceiver *fc_tqueue = (TQueueDestReceiver *) fc_self;
	MinimalTuple fc_tuple;
	shm_mq_result fc_result;
	bool		fc_should_free;

	/* 发送元组本身。 */
	fc_tuple = ExecFetchSlotMinimalTuple(fc_slot, &fc_should_free);
	fc_result = shm_mq_send(fc_tqueue->queue, fc_tuple->t_len, fc_tuple, false, false);

	if (fc_should_free)
		pfree(fc_tuple);

	/* 检查失败。 */
	if (fc_result == SHM_MQ_DETACHED)
		return false;
	else if (fc_result != SHM_MQ_SUCCESS)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("could not send tuple to shared-memory queue")));

	return true;
}

/*
 * 准备接收来自执行器的元组。
 */
static void fc_tqueueStartupReceiver(DestReceiver *fc_self, int fc_operation, TupleDesc fc_typeinfo)
{
	/* 不执行任何操作 */
}

/*
 * 在执行器运行结束时进行清理
 */
static void fc_tqueueShutdownReceiver(DestReceiver *fc_self)
{
	TQueueDestReceiver *fc_tqueue = (TQueueDestReceiver *) fc_self;

	if (fc_tqueue->queue != NULL)
		shm_mq_detach(fc_tqueue->queue);
	fc_tqueue->queue = NULL;
}

/*
 * 完成后销毁接收器
 */
static void fc_tqueueDestroyReceiver(DestReceiver *fc_self)
{
	TQueueDestReceiver *fc_tqueue = (TQueueDestReceiver *) fc_self;

	/* 我们可能已经从队列中分离，但还是确认一下 */
	if (fc_tqueue->queue != NULL)
		shm_mq_detach(fc_tqueue->queue);
	pfree(fc_self);
}

/*
 * 创建一个 DestReceiver，将元组写入元组队列。
 */
DestReceiver * CreateTupleQueueDestReceiver(shm_mq_handle *fc_handle)
{
	TQueueDestReceiver *fc_self;

	fc_self = (TQueueDestReceiver *) palloc0(sizeof(TQueueDestReceiver));

	fc_self->pub.receiveSlot = fc_tqueueReceiveSlot;
	fc_self->pub.rStartup = fc_tqueueStartupReceiver;
	fc_self->pub.rShutdown = fc_tqueueShutdownReceiver;
	fc_self->pub.rDestroy = fc_tqueueDestroyReceiver;
	fc_self->pub.mydest = DestTupleQueue;
	fc_self->queue = fc_handle;

	return (DestReceiver *) fc_self;
}

/*
 * 创建一个元组队列读取器。
 */
TupleQueueReader *
CreateTupleQueueReader(shm_mq_handle *fc_handle)
{
	TupleQueueReader *fc_reader = palloc0(sizeof(TupleQueueReader));

	fc_reader->queue = fc_handle;

	return fc_reader;
}

/*
 * 销毁元组队列读取器。
 *
 * 注意：清理底层的 shm_mq 是调用者的责任。
 * 我们在这里不会访问它，因为它可能已经被分离。
 */
void DestroyTupleQueueReader(TupleQueueReader *fc_reader)
{
	pfree(fc_reader);
}

/*
 * 从元组队列读取器中获取一个元组。
 *
 * 如果没有剩余的元组，或者如果 nowait = true 且没有元组准备好返回，返回值为 NULL。 *done，如果不为 NULL，
 * 在没有剩余元组时设置为 true，否则设置为 false。
 *
 * 返回的元组（如果有的话）要么在共享内存中，要么在私人缓冲区中
 * 不应释放。 在下一次调用 TupleQueueReaderNext() 后，指针无效。
 *
 * 即使 shm_mq_receive() 返回 SHM_MQ_WOULD_BLOCK，这仍然可以
 * 从部分读取的消息中积累字节，因此即使没有返回任何内容，调用时将 nowait 设置为 true 也是有用的。
 */
MinimalTuple TupleQueueReaderNext(TupleQueueReader *fc_reader, bool fc_nowait, bool *fc_done)
{
	MinimalTuple fc_tuple;
	shm_mq_result fc_result;
	Size		fc_nbytes;
	void	   *fc_data;

	if (fc_done != NULL)
		*fc_done = false;

	/* 尝试读取消息。 */
	fc_result = shm_mq_receive(fc_reader->queue, &fc_nbytes, &fc_data, fc_nowait);

	/* 如果队列已分离，设置 *done 并返回 NULL。 */
	if (fc_result == SHM_MQ_DETACHED)
	{
		if (fc_done != NULL)
			*fc_done = true;
		return NULL;
	}

	/* 在非阻塞模式下，如果尚未准备好消息则退出。 */
	if (fc_result == SHM_MQ_WOULD_BLOCK)
		return NULL;
	Assert(fc_result == SHM_MQ_SUCCESS);

	/*
	 * 直接返回一个指向队列内存的指针（最好是
	 * 足够对齐的）。
	 */
	fc_tuple = (MinimalTuple) fc_data;
	Assert(fc_tuple->t_len == fc_nbytes);

	return fc_tuple;
}
